package com.atguigu.flink.day08;

import com.atguigu.flink.beans.WaterSensor;
import com.atguigu.flink.func.WaterSensorMapFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

/**
 * @author Felix
 * @date 2023/12/9
 * 该案例演示了状态的TimeToLive
 */
public class Flink06_KeyedState_TTL {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        SingleOutputStreamOperator<WaterSensor> wsDS = env
            .socketTextStream("hadoop102", 8888)
            .map(new WaterSensorMapFunction());

        //按照传感器的id进行分组
        KeyedStream<WaterSensor, String> keyedDS = wsDS.keyBy(WaterSensor::getId);

        //对分组后的数据进行处理
        keyedDS.process(
            new KeyedProcessFunction<String, WaterSensor, String>() {
                //声明状态，用于记录"当前"传感器上次采集到的水位值
                private ValueState<Integer> lastVcState ;

                // 不能在状态声明的时候直接进行初始化，因为这个时候获取不到RuntimeContext
                // private ValueState<Integer> lastVcState  = getRuntimeContext().getState(new ValueStateDescriptor<Integer>("lastVcState", Integer.class));

                //给状态进行初始化:在生命周期开始的时候
                @Override
                public void open(Configuration parameters) throws Exception {
                    //定义状态描述器
                    //参数1：状态标记名称
                    //参数2：值状态中存放的元素类型
                    ValueStateDescriptor<Integer> valueStateDescriptor
                        = new ValueStateDescriptor<Integer>("lastVcState", Integer.class);

                    //设置当前状态失效时间  注意：时间指的是处理时间
                    valueStateDescriptor.enableTimeToLive(StateTtlConfig.newBuilder(Time.seconds(10))
                        // .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
                        // .setUpdateType(StateTtlConfig.UpdateType.OnReadAndWrite)
                        // .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
                        .build());

                    lastVcState = getRuntimeContext().getState(valueStateDescriptor);
                }

                @Override
                public void processElement(WaterSensor ws, Context ctx, Collector<String> out) throws Exception {
                    //获取本次水位值
                    Integer vc = ws.vc;
                    //获取上次水位值    .value()从状态中获取存储的值
                    Integer lastVc = lastVcState.value();
                    lastVc = lastVc == null ? 0 : lastVc;
                    if (Math.abs(vc - lastVc) > 10) {
                        out.collect("传感器id：" + ws.id + ",当前水位值" + vc + ",和上一次水位值" + lastVc + "相差大于10~");
                    }
                    //.update  将参数中给定的值  更新到状态中

                    if(vc > 10){
                        lastVcState.update(vc);
                    }

                    // 清空状态
                    // lastVcState.clear();
                }
            }
        ).print();
        /*
        //错误思路：使用普通的变量接收上次采集到的水位值，因为一个并行度上，
        可能有多个分组，多个分组共享同一个普通的成员变量
        keyedDS.process(
            new KeyedProcessFunction<String, WaterSensor, String>() {
                //用于记录"当前"传感器上次采集到的水位值
                Integer lastVc = 0;
                @Override
                public void processElement(WaterSensor ws, Context ctx, Collector<String> out) throws Exception {
                    //获取当前传感器本次采集到的水位值
                    Integer vc = ws.vc;
                    if(Math.abs(vc - lastVc) > 10){
                        out.collect("传感器id：" + ws.id + ",当前水位值" + vc+",和上一次水位值" + lastVc+"相差大于10~");
                    }
                    lastVc = vc;
                }
            }
        ).print();*/

        env.execute();
    }
}
